package pipeline

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"fmt"
	"os"
	"path/filepath"
	"sync"
	"time"

	"mcp-server/internal/docprocessor"
	"mcp-server/internal/metadata"
	"mcp-server/internal/storage"
	"mcp-server/internal/vector"
)

// PipelineManager 流水线管理器
type PipelineManager struct {
	config            *PipelineConfig
	processors        []ProcessorInterface
	progressMonitor   ProgressMonitor
	checkpointManager CheckpointManager

	// 组件依赖
	docProcessor    *docprocessor.DocumentProcessor
	metadataManager *metadata.DefaultManager
	vectorManager   *vector.VectorManager
	storageManager  *storage.StorageManager

	// 运行时状态
	tasks     map[string]*PipelineTask
	taskQueue chan *PipelineTask
	workers   []*Worker
	metrics   *PipelineMetrics

	// 同步控制
	mutex   sync.RWMutex
	running bool
	ctx     context.Context
	cancel  context.CancelFunc
	wg      sync.WaitGroup
}

// Worker 工作器
type Worker struct {
	id       int
	manager  *PipelineManager
	taskChan chan *PipelineTask
	ctx      context.Context
	cancel   context.CancelFunc
}

// NewPipelineManager 创建流水线管理器
func NewPipelineManager(
	config *PipelineConfig,
	docProcessor *docprocessor.DocumentProcessor,
	metadataManager *metadata.DefaultManager,
	vectorManager *vector.VectorManager,
	storageManager *storage.StorageManager,
) *PipelineManager {
	if config == nil {
		config = DefaultPipelineConfig()
	}

	ctx, cancel := context.WithCancel(context.Background())

	manager := &PipelineManager{
		config:          config,
		docProcessor:    docProcessor,
		metadataManager: metadataManager,
		vectorManager:   vectorManager,
		storageManager:  storageManager,
		tasks:           make(map[string]*PipelineTask),
		taskQueue:       make(chan *PipelineTask, config.QueueSize),
		metrics:         &PipelineMetrics{LastUpdateTime: time.Now()},
		ctx:             ctx,
		cancel:          cancel,
	}

	// 初始化组件
	manager.progressMonitor = NewProgressMonitor()
	manager.checkpointManager = NewCheckpointManager(config.CheckpointPath)
	manager.initializeProcessors()

	return manager
}

// initializeProcessors 初始化处理器
func (pm *PipelineManager) initializeProcessors() {
	pm.processors = []ProcessorInterface{
		NewFileReaderProcessor(),
		NewDocumentParserProcessor(pm.docProcessor),
		NewMetadataExtractorProcessor(pm.metadataManager),
		NewTextSplitterProcessor(pm.docProcessor),
		NewVectorizerProcessor(pm.vectorManager),
		NewStorageProcessor(pm.storageManager),
	}
}

// Start 启动流水线
func (pm *PipelineManager) Start() error {
	pm.mutex.Lock()
	defer pm.mutex.Unlock()

	if pm.running {
		return fmt.Errorf("pipeline is already running")
	}

	// 创建工作器
	pm.workers = make([]*Worker, pm.config.WorkerPoolSize)
	for i := 0; i < pm.config.WorkerPoolSize; i++ {
		worker := &Worker{
			id:       i,
			manager:  pm,
			taskChan: make(chan *PipelineTask, 1),
		}
		worker.ctx, worker.cancel = context.WithCancel(pm.ctx)
		pm.workers[i] = worker

		pm.wg.Add(1)
		go worker.run()
	}

	// 启动任务分发器
	pm.wg.Add(1)
	go pm.taskDispatcher()

	// 启动指标收集器
	if pm.config.MetricsEnabled {
		pm.wg.Add(1)
		go pm.metricsCollector()
	}

	// 启动检查点管理器
	if pm.config.CheckpointEnabled {
		pm.wg.Add(1)
		go pm.checkpointWorker()
	}

	pm.running = true
	return nil
}

// generateTaskID 生成任务ID
func generateTaskID() string {
	bytes := make([]byte, 16)
	rand.Read(bytes)
	return hex.EncodeToString(bytes)
}

// Stop 停止流水线
func (pm *PipelineManager) Stop() error {
	pm.mutex.Lock()
	defer pm.mutex.Unlock()

	if !pm.running {
		return fmt.Errorf("pipeline is not running")
	}

	// 取消上下文
	pm.cancel()

	// 停止所有工作器
	for _, worker := range pm.workers {
		worker.cancel()
	}

	// 等待所有goroutine完成
	pm.wg.Wait()

	pm.running = false
	return nil
}

// IsRunning 检查是否运行中
func (pm *PipelineManager) IsRunning() bool {
	pm.mutex.RLock()
	defer pm.mutex.RUnlock()
	return pm.running
}

// SubmitTask 提交任务
func (pm *PipelineManager) SubmitTask(filePath string, config *TaskConfig) (string, error) {
	if !pm.IsRunning() {
		return "", fmt.Errorf("pipeline is not running")
	}

	// 检查文件是否存在
	if _, err := os.Stat(filePath); os.IsNotExist(err) {
		return "", fmt.Errorf("file does not exist: %s", filePath)
	}

	// 使用默认配置
	if config == nil {
		config = pm.config.DefaultTaskConfig
	}

	// 创建任务
	taskID := generateTaskID()
	ctx, cancel := context.WithTimeout(pm.ctx, pm.config.TaskTimeout)

	task := &PipelineTask{
		ID:         taskID,
		FilePath:   filePath,
		FileName:   filepath.Base(filePath),
		Config:     config,
		Context:    ctx,
		CancelFunc: cancel,
		CreatedAt:  time.Now(),
		UpdatedAt:  time.Now(),
		Progress: &TaskProgress{
			TaskID:     taskID,
			FileName:   filepath.Base(filePath),
			Stage:      StageQueued,
			Status:     StatusPending,
			Progress:   0,
			StartTime:  time.Now(),
			UpdateTime: time.Now(),
		},
	}

	// 添加到任务列表
	pm.mutex.Lock()
	pm.tasks[taskID] = task
	pm.metrics.TotalTasks++
	pm.metrics.QueuedTasks++
	pm.mutex.Unlock()

	// 提交到队列
	select {
	case pm.taskQueue <- task:
		return taskID, nil
	case <-pm.ctx.Done():
		return "", fmt.Errorf("pipeline is shutting down")
	default:
		return "", fmt.Errorf("task queue is full")
	}
}

// SubmitBatch 批量提交任务
func (pm *PipelineManager) SubmitBatch(filePaths []string, config *TaskConfig) ([]string, error) {
	var taskIDs []string
	var errors []string

	for _, filePath := range filePaths {
		taskID, err := pm.SubmitTask(filePath, config)
		if err != nil {
			errors = append(errors, fmt.Sprintf("%s: %v", filePath, err))
		} else {
			taskIDs = append(taskIDs, taskID)
		}
	}

	if len(errors) > 0 {
		return taskIDs, fmt.Errorf("some tasks failed to submit: %v", errors)
	}

	return taskIDs, nil
}

// CancelTask 取消任务
func (pm *PipelineManager) CancelTask(taskID string) error {
	pm.mutex.RLock()
	task, exists := pm.tasks[taskID]
	pm.mutex.RUnlock()

	if !exists {
		return fmt.Errorf("task not found: %s", taskID)
	}

	task.CancelFunc()
	task.SetStatus(StatusCancelled)

	return nil
}

// GetTask 获取任务
func (pm *PipelineManager) GetTask(taskID string) (*PipelineTask, error) {
	pm.mutex.RLock()
	defer pm.mutex.RUnlock()

	task, exists := pm.tasks[taskID]
	if !exists {
		return nil, fmt.Errorf("task not found: %s", taskID)
	}

	return task, nil
}

// ListTasks 列出所有任务
func (pm *PipelineManager) ListTasks() ([]*PipelineTask, error) {
	pm.mutex.RLock()
	defer pm.mutex.RUnlock()

	tasks := make([]*PipelineTask, 0, len(pm.tasks))
	for _, task := range pm.tasks {
		tasks = append(tasks, task)
	}

	return tasks, nil
}

// GetProgress 获取任务进度
func (pm *PipelineManager) GetProgress(taskID string) (*TaskProgress, error) {
	return pm.progressMonitor.GetProgress(taskID)
}

// GetMetrics 获取流水线指标
func (pm *PipelineManager) GetMetrics() *PipelineMetrics {
	pm.mutex.RLock()
	defer pm.mutex.RUnlock()

	// 返回指标的副本
	metrics := *pm.metrics
	return &metrics
}

// UpdateConfig 更新配置
func (pm *PipelineManager) UpdateConfig(config *PipelineConfig) error {
	pm.mutex.Lock()
	defer pm.mutex.Unlock()

	pm.config = config
	return nil
}

// GetConfig 获取配置
func (pm *PipelineManager) GetConfig() *PipelineConfig {
	pm.mutex.RLock()
	defer pm.mutex.RUnlock()

	// 返回配置的副本
	config := *pm.config
	return &config
}

// taskDispatcher 任务分发器
func (pm *PipelineManager) taskDispatcher() {
	defer pm.wg.Done()

	for {
		select {
		case task := <-pm.taskQueue:
			// 找到空闲的工作器
			assigned := false
			for _, worker := range pm.workers {
				select {
				case worker.taskChan <- task:
					assigned = true
					pm.mutex.Lock()
					pm.metrics.QueuedTasks--
					pm.metrics.ActiveTasks++
					pm.mutex.Unlock()
					break
				default:
					continue
				}
				if assigned {
					break
				}
			}

			// 如果没有空闲工作器，重新放回队列
			if !assigned {
				select {
				case pm.taskQueue <- task:
				case <-pm.ctx.Done():
					return
				}
			}

		case <-pm.ctx.Done():
			return
		}
	}
}

// metricsCollector 指标收集器
func (pm *PipelineManager) metricsCollector() {
	defer pm.wg.Done()

	ticker := time.NewTicker(pm.config.ProgressUpdateInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			pm.updateMetrics()
		case <-pm.ctx.Done():
			return
		}
	}
}

// updateMetrics 更新指标
func (pm *PipelineManager) updateMetrics() {
	pm.mutex.Lock()
	defer pm.mutex.Unlock()

	now := time.Now()
	duration := now.Sub(pm.metrics.LastUpdateTime)

	// 计算吞吐量
	if duration > 0 {
		completedInPeriod := pm.metrics.CompletedTasks
		pm.metrics.ThroughputPerMin = float64(completedInPeriod) / duration.Minutes()
	}

	// 计算错误率
	if pm.metrics.TotalTasks > 0 {
		pm.metrics.ErrorRate = float64(pm.metrics.FailedTasks) / float64(pm.metrics.TotalTasks)
	}

	pm.metrics.LastUpdateTime = now
}

// checkpointWorker 检查点工作器
func (pm *PipelineManager) checkpointWorker() {
	defer pm.wg.Done()

	ticker := time.NewTicker(pm.config.CheckpointInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			pm.saveCheckpoints()
		case <-pm.ctx.Done():
			return
		}
	}
}

// saveCheckpoints 保存检查点
func (pm *PipelineManager) saveCheckpoints() {
	pm.mutex.RLock()
	tasks := make([]*PipelineTask, 0, len(pm.tasks))
	for _, task := range pm.tasks {
		if task.Progress.Status == StatusProcessing {
			tasks = append(tasks, task)
		}
	}
	pm.mutex.RUnlock()

	for _, task := range tasks {
		if err := pm.checkpointManager.SaveCheckpoint(task); err != nil {
			// 记录错误但不中断处理
			fmt.Printf("Failed to save checkpoint for task %s: %v\n", task.ID, err)
		}
	}
}

// Worker.run 工作器运行
func (w *Worker) run() {
	defer w.manager.wg.Done()

	for {
		select {
		case task := <-w.taskChan:
			w.processTask(task)
		case <-w.ctx.Done():
			return
		}
	}
}

// processTask 处理任务
func (w *Worker) processTask(task *PipelineTask) {
	startTime := time.Now()
	task.SetStatus(StatusProcessing)
	task.Progress.StartTime = startTime

	// 更新进度监控
	w.manager.progressMonitor.UpdateProgress(task.ID, task.GetProgress())

	// 执行处理流程
	err := w.executeProcessors(task)

	// 更新任务状态
	endTime := time.Now()
	processingTime := endTime.Sub(startTime)

	if err != nil {
		task.SetError(err)
		w.manager.mutex.Lock()
		w.manager.metrics.FailedTasks++
		w.manager.mutex.Unlock()
	} else {
		task.SetStatus(StatusCompleted)
		task.UpdateProgress(StageCompleted, 100)
		w.manager.mutex.Lock()
		w.manager.metrics.CompletedTasks++
		w.manager.mutex.Unlock()

		// 设置结果
		if task.Result == nil {
			task.Result = &TaskResult{}
		}
		task.Result.Success = true
		task.Result.ProcessingTime = processingTime
	}

	// 更新指标
	w.manager.mutex.Lock()
	w.manager.metrics.ActiveTasks--
	w.manager.mutex.Unlock()

	// 最终进度更新
	w.manager.progressMonitor.UpdateProgress(task.ID, task.GetProgress())

	// 清理检查点
	if w.manager.config.CheckpointEnabled {
		w.manager.checkpointManager.DeleteCheckpoint(task.ID)
	}
}

// executeProcessors 执行处理器
func (w *Worker) executeProcessors(task *PipelineTask) error {
	totalStages := len(w.manager.processors)

	for i, processor := range w.manager.processors {
		// 检查是否取消
		select {
		case <-task.Context.Done():
			return fmt.Errorf("task cancelled")
		default:
		}

		// 更新进度
		progress := float64(i) / float64(totalStages) * 100
		task.UpdateProgress(processor.GetStage(), progress)
		w.manager.progressMonitor.UpdateProgress(task.ID, task.GetProgress())

		// 执行处理器
		stageCtx, cancel := context.WithTimeout(task.Context, w.manager.config.StageTimeout)
		err := processor.Process(stageCtx, task)
		cancel()

		if err != nil {
			return fmt.Errorf("stage %s failed: %w", processor.GetStage(), err)
		}
	}

	return nil
}
